package main

import (
	"fmt"
	"log"
	"math"
	"math/rand"
	"os"
	"sort"
	"sync"
	"time"
)

type ISchedule interface {
	Schedule()
	Finish()
}

func NewSchedule(name string, record *Record, finishChan chan Task) ISchedule {
	if name == "affinity" {
		return newAffinitySchedule(record, finishChan)
	}
	return newRandSchedule(record, finishChan)
}

type BaseSchedule struct {
	mu        sync.Mutex
	record    *Record
	waitTasks []*Task
	assignMap map[string]*Task

	finishChan chan Task
}

func (r *BaseSchedule) Schedule() {
	log.Fatal("Can not use it")
}

func (r *BaseSchedule) Finish() {
	file, err := os.OpenFile("./res.log", os.O_RDWR|os.O_APPEND|os.O_CREATE, 0644)
	if err != nil {
		log.Println("Open file error")
		return
	}
	defer file.Close()

	for {
		item := <-r.finishChan

		r.mu.Lock()
		real, ok := r.assignMap[item.UUID]
		log.Printf("The finish task is %v\n", item.UUID)
		if !ok {
			log.Printf("Can not get task with uuid %v.\n", item.UUID)
			continue
		}
		delete(r.assignMap, item.UUID)
		r.mu.Unlock()

		real.FinishTime = item.FinishTime
		real.ReceiveTime = item.ReceiveTime
		real.StartTime = item.StartTime
		real.ReadyTime = item.ReadyTime

		tmp := fmt.Sprintf("UUID: %v, Submit Time: %v, Commit Time: %v, Ready Time: %v, Arrive Time: %v, Start Time: %v, Finish Time: %v, Data Size: %vKB\n",
			real.UUID,
			time.Unix(real.CommitTime, 0).Format("2006-01-02 15:04:05"),
			time.Unix(real.AssignTime, 0).Format("2006-01-02 15:04:05"),
			time.Unix(real.ReceiveTime, 0).Format("2006-01-02 15:04:05"),
			time.Unix(real.ReadyTime, 0).Format("2006-01-02 15:04:05"),
			time.Unix(real.StartTime, 0).Format("2006-01-02 15:04:05"),
			time.Unix(real.FinishTime, 0).Format("2006-01-02 15:04:05"),
			real.Position.ID,
		)
		n, err := file.WriteString(tmp)
		log.Printf("Write %v char\n", n)
		if err != nil {
			log.Println(err.Error())
		}
	}
}

type RandSchedule struct {
	BaseSchedule
}

func newRandSchedule(record *Record, finishChan chan Task) *RandSchedule {
	return &RandSchedule{
		BaseSchedule: BaseSchedule{
			record:     record,
			waitTasks:  []*Task{},
			assignMap:  map[string]*Task{},
			finishChan: finishChan,
		},
	}
}

func (r *RandSchedule) scheduler(done chan struct{}) {
	for {
		select {
		// 调度器拥有执行窗口到时间退出
		case <-done:
			return
		default:
			if res := r.mu.TryLock(); !res {
				continue
			}

			if res := r.record.mu.TryLock(); !res {
				continue
			}

			ks := []string{}
			for k := range r.record.followerMap {
				ks = append(ks, k)

				// copy task
				r.waitTasks = append(r.waitTasks, r.record.followerMap[k].task...)
				r.record.followerMap[k].task = []*Task{}
			}
			if len(r.waitTasks) == 0 {
				r.record.mu.Unlock()
				r.mu.Unlock()
				continue
			}

			sort.Strings(ks)

			idx := rand.Intn(len(ks))
			ip := ks[idx]

			item := r.waitTasks[0]
			err := CommitTask(ip, item)
			item.AssignTime = time.Now().Unix()
			// record 解锁
			r.record.mu.Unlock()

			if err != nil {
				// task 解锁
				r.mu.Unlock()
				continue
			}
			// 从等待队列中删除
			r.assignMap[item.UUID] = item
			r.waitTasks = r.waitTasks[1:]

			// task 解锁
			r.mu.Unlock()
		}
	}
}

func (r *RandSchedule) Schedule() {
	rand.Seed(time.Now().Unix())

	// 守护调度器执行
	for {
		// 指定一个1s的执行窗口
		done := make(chan struct{})
		go r.scheduler(done)

		timer := time.NewTimer(1 * time.Second)
		<-timer.C
		done <- struct{}{}
		close(done)

		time.Sleep(5 * time.Second)
	}
}

type AffinitySchedule struct {
	BaseSchedule
}

func newAffinitySchedule(record *Record, finishChan chan Task) *AffinitySchedule {
	return &AffinitySchedule{
		BaseSchedule: BaseSchedule{
			record:     record,
			waitTasks:  []*Task{},
			assignMap:  map[string]*Task{},
			finishChan: finishChan,
		},
	}
}

type affinity struct {
	ip     string
	alpha  float64
	beta   float64
	lpcomp float64
	lpstor float64

	res float64
}

func getCos(x, y []int) float64 {
	upper := 0.0
	downX := 0.0
	downY := 0.0

	for i := 0; i < len(x); i++ {
		upper += float64(x[i] * y[i])
		downX += float64(x[i] * x[i])
		downY += float64(y[i] * y[i])
	}

	return upper / (math.Pow(downX, 0.5) * math.Pow(downY, 0.5))
}

func (r *AffinitySchedule) scheduler() {
	r.mu.Lock()
	r.record.mu.Lock()

	for k := range r.record.followerMap {
		// copy task
		r.waitTasks = append(r.waitTasks, r.record.followerMap[k].task...)
		r.record.followerMap[k].task = []*Task{}
	}

	if len(r.waitTasks) == 0 {
		r.record.mu.Unlock()
		r.mu.Unlock()
		return
	}

	item := r.waitTasks[0]
	tmpNode := []affinity{}
	// 初步筛查
	for _, v := range r.record.followerMap {
		if item.CPU > int(v.cpuMax) || item.Storage > int(v.storageTotal)-int(v.storageUsing) || item.Time > v.time {
			continue
		}
		// 计算余弦相似度
		alpha := getCos([]int{item.CPU, item.Storage}, []int{int(v.cpuTotal), int(v.storageTotal)})
		tmpNode = append(tmpNode, affinity{
			ip:    v.ip,
			alpha: alpha,
			// 获取计算和存储资源负载
			lpcomp: float64(v.cpuUsing+v.cpuWait+int32(item.CPU)) / float64(v.cpuTotal),
			lpstor: float64(v.storageUsing+int32(item.Storage)) / float64(v.storageTotal),
		})
	}
	if len(tmpNode) == 0 {
		r.record.mu.Unlock()
		r.mu.Unlock()
		return
	}

	// 归一化lpcomp和lpstor
	lpcompMax, lpcompMin := 0.0, math.MaxFloat64
	lpstorMax, lpstorMin := 0.0, math.MaxFloat64
	for _, node := range tmpNode {
		if node.lpcomp > lpcompMax {
			lpcompMax = node.lpcomp
		}
		if node.lpcomp < lpcompMin {
			lpcompMin = node.lpcomp
		}
		if node.lpstor > lpstorMax {
			lpstorMax = node.lpstor
		}
		if node.lpstor < lpstorMin {
			lpstorMin = node.lpstor
		}
	}

	lambda := 0.5
	for i, node := range tmpNode {
		if lpcompMax == lpcompMin {
			tmpNode[i].lpcomp = 1
		} else {
			tmpNode[i].lpcomp = (node.lpcomp - lpcompMin) / (lpcompMax - lpcompMin)
		}
		if lpstorMax == lpstorMin {
			tmpNode[i].lpstor = 1
		} else {
			tmpNode[i].lpstor = (node.lpstor - lpstorMin) / (lpstorMax - lpstorMin)
		}

		tmpNode[i].beta = math.Pow((1-tmpNode[i].lpcomp)*(1-tmpNode[i].lpcomp)+(1-tmpNode[i].lpstor)*(1-tmpNode[i].lpstor), 0.5)
		tmpNode[i].res = lambda*((tmpNode[i].alpha+tmpNode[i].beta)/2) + (1-lambda)*math.Pow(tmpNode[i].alpha*tmpNode[i].beta, 0.5)
	}

	sort.Slice(tmpNode, func(i, j int) bool {
		return tmpNode[i].res > tmpNode[j].res
	})
	ip := tmpNode[0].ip

	item.AssignTime = time.Now().Unix()
	err := CommitTask(ip, item)
	// record 解锁
	r.record.mu.Unlock()

	if err != nil {
		// task 解锁
		r.mu.Unlock()
		return
	}
	log.Printf("Assign task to %v\n", ip)
	// 从等待队列中删除
	r.assignMap[item.UUID] = item
	r.waitTasks = r.waitTasks[1:]

	// task 解锁
	r.mu.Unlock()

}

// func (r *AffinitySchedule) scheduler(done chan struct{}) {
// 	for {
// 		select {
// 		// 调度器拥有执行窗口到时间退出
// 		case <-done:
// 			return
// 		default:
// 			r.mu.Lock()
// 			r.record.mu.Lock()

// 			for k := range r.record.followerMap {
// 				// copy task
// 				r.waitTasks = append(r.waitTasks, r.record.followerMap[k].task...)
// 				r.record.followerMap[k].task = []*Task{}
// 			}

// 			if len(r.waitTasks) == 0 {
// 				r.record.mu.Unlock()
// 				r.mu.Unlock()
// 				continue
// 			}

// 			item := r.waitTasks[0]
// 			tmpNode := []affinity{}
// 			// 初步筛查
// 			for _, v := range r.record.followerMap {
// 				if item.CPU > int(v.cpuMax) || item.Storage > int(v.storageTotal)-int(v.storageUsing) || item.Time > v.time {
// 					continue
// 				}
// 				// 计算余弦相似度
// 				alpha := getCos([]int{item.CPU, item.Storage}, []int{int(v.cpuTotal), int(v.storageTotal)})
// 				tmpNode = append(tmpNode, affinity{
// 					ip:    v.ip,
// 					alpha: alpha,
// 					// 获取计算和存储资源负载
// 					lpcomp: float64(v.cpuUsing+v.cpuWait+int32(item.CPU)) / float64(v.cpuTotal),
// 					lpstor: float64(v.storageUsing+int32(item.Storage)) / float64(v.storageTotal),
// 				})
// 			}
// 			if len(tmpNode) == 0 {
// 				r.record.mu.Unlock()
// 				r.mu.Unlock()
// 				continue
// 			}

// 			// 归一化lpcomp和lpstor
// 			lpcompMax, lpcompMin := 0.0, math.MaxFloat64
// 			lpstorMax, lpstorMin := 0.0, math.MaxFloat64
// 			for _, node := range tmpNode {
// 				if node.lpcomp > lpcompMax {
// 					lpcompMax = node.lpcomp
// 				}
// 				if node.lpcomp < lpcompMin {
// 					lpcompMin = node.lpcomp
// 				}
// 				if node.lpstor > lpstorMax {
// 					lpstorMax = node.lpstor
// 				}
// 				if node.lpstor < lpstorMin {
// 					lpstorMin = node.lpstor
// 				}
// 			}

// 			lambda := 0.5
// 			for i, node := range tmpNode {
// 				if lpcompMax == lpcompMin {
// 					tmpNode[i].lpcomp = 1
// 				} else {
// 					tmpNode[i].lpcomp = (node.lpcomp - lpcompMin) / (lpcompMax - lpcompMin)
// 				}
// 				if lpstorMax == lpstorMin {
// 					tmpNode[i].lpstor = 1
// 				} else {
// 					tmpNode[i].lpstor = (node.lpstor - lpstorMin) / (lpstorMax - lpstorMin)
// 				}

// 				tmpNode[i].beta = math.Pow((1-tmpNode[i].lpcomp)*(1-tmpNode[i].lpcomp)+(1-tmpNode[i].lpstor)*(1-tmpNode[i].lpstor), 0.5)
// 				tmpNode[i].res = lambda*((tmpNode[i].alpha+tmpNode[i].beta)/2) + (1-lambda)*math.Pow(tmpNode[i].alpha*tmpNode[i].beta, 0.5)
// 			}

// 			sort.Slice(tmpNode, func(i, j int) bool {
// 				return tmpNode[i].res > tmpNode[j].res
// 			})
// 			ip := tmpNode[0].ip

// 			item.AssignTime = time.Now().Unix()
// 			err := CommitTask(ip, item)
// 			// record 解锁
// 			r.record.mu.Unlock()

// 			if err != nil {
// 				// task 解锁
// 				r.mu.Unlock()
// 				continue
// 			}
// 			log.Printf("Assign task to %v\n", ip)
// 			// 从等待队列中删除
// 			r.assignMap[item.UUID] = item
// 			r.waitTasks = r.waitTasks[1:]

// 			// task 解锁
// 			r.mu.Unlock()
// 		}

// 		time.Sleep(1500 * time.Millisecond)
// 	}
// }

func (r *AffinitySchedule) Schedule() {
	// 守护调度器执行
	for {
		r.scheduler()

		time.Sleep(1500 * time.Millisecond)
	}
}

// func (r *AffinitySchedule) Schedule() {
// 	rand.Seed(time.Now().Unix())

// 	// 守护调度器执行
// 	for {
// 		// 指定一个1s的执行窗口
// 		done := make(chan struct{})
// 		go r.scheduler(done)

// 		timer := time.NewTimer(1 * time.Second)
// 		<-timer.C
// 		done <- struct{}{}
// 		close(done)

// 		time.Sleep(5 * time.Second)
// 	}
// }
